﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Lucene.Net.Index;
using Apache.Cassandra;
using Lucene.Net.Documents;
using Lucandra.Convenience;
using Lucene.Net.Search;
using Lucandra.Net.Connection.Pooling;
using Lucandra.Net.Connection;
using Lucandra.Configuration;
using System.Threading;

namespace Lucandra.Net
{
    public sealed class LucandraReader : IndexReader
    {
        #region Constants

        private const int MaximumDocuments = 1000000;

        #endregion

        #region Properties [public]

        public string IndexName { get; private set; }
        public string Keyspace { get; private set; }

        #endregion
        #region Properties [private]

        private Dictionary<int, LucandraDocument> DocumentCache { get; set; }
        private Dictionary<string, int> DocumentIDToDocumentNumberMap { get; set; }
        private Dictionary<int, string> DocumentNumberToDocumentIDMap { get; set; }
        private int DocumentCounter { get; set; }
        private Dictionary<string, byte[]> FieldNorms { get; set; }

        #endregion
        #region Properties [internal]

        #endregion

        #region Constructors

        /// <summary>
        /// Creates a new LucandraReader on the specified index.
        /// </summary>
        /// <param name="indexName">The name of the index to open this reader for.</param>
        public LucandraReader(string indexName)
            : base()
        {
            this.IndexName = indexName;
            this.DocumentCache = new Dictionary<int, LucandraDocument>();
            this.DocumentIDToDocumentNumberMap = new Dictionary<string, int>();
            this.DocumentNumberToDocumentIDMap = new Dictionary<int, string>();
            this.DocumentCounter = 0;
            this.FieldNorms = new Dictionary<string, byte[]>();
            this.Keyspace = KeyspaceManager.GetKeyspaceFromIndexName(indexName);
        }

        /// <summary>
        /// Creates a new LucandraReader on the default index specified in your app/web configuration file.
        /// </summary>
        public LucandraReader()
            : this(LucandraConfigurationSection.GetConfig().DefaultIndexName)
        {
        }

        #endregion

        #region IndexReader Implementations

        /// <summary>
        /// Gets the term frequency vectors for the specified document.
        /// </summary>
        /// <param name="docNumber">The document number as known by this IndexReader.</param>
        /// <returns>An array of term frequency vectors for the specified document.</returns>
        public override TermFreqVector[] GetTermFreqVectors(int docNumber)
        {
            // Get the cached LucandraDocument.
            var lucandraDocument = this.DocumentCache[docNumber];

            var list = new List<TermFreqVector>();

            // Determine all field names in the indexed terms.
            string[] fieldNames = (from x in lucandraDocument.IndexedTerms select x.FieldName).Distinct().ToArray();
            
            // Iterate through each field and get the terms/frequencies for each.  Append the LucandraTermVector
            // for each field.
            foreach (string fieldName in fieldNames)
            {
                string[] terms = lucandraDocument.IndexedTerms.Where(t => t.FieldName == fieldName).Select(t => t.TermText).ToArray();
                int[] frequencies = lucandraDocument.IndexedTerms.Where(t => t.FieldName == fieldName).Select(t => t.Frequency).ToArray();
                list.Add(new LucandraTermVector(fieldName, terms, frequencies));
            }

            return list.ToArray();
        }

        /// <summary>
        /// Gets the term frequency vector for the specified document and field.  The returned vector contains terms and frequencies for the
        /// terms in the specified field of this document, if the field had the StoreTermVector flag set.  If term vectors had been stored
        /// with positions or offsets, a LucandraTermPositionVector is returned.
        /// </summary>
        /// <param name="docNumber">The document number as known by this IndexReader.</param>
        /// <param name="field">The field for which the term frequency vector is returned.</param>
        /// <returns>The LucandraTermVector for the specified document and field, or a LucandraTermPositionVector if vectors were stored
        /// with positions or offsets.</returns>
        public override TermFreqVector GetTermFreqVector(int docNumber, string field)
        {
            var lucandraDocument = this.GetAndCacheLucandraDocument(docNumber);
            if (lucandraDocument == null)
                return null;

            var termVector = LucandraTermVector.FromLucandraDocument(lucandraDocument, field);
            return termVector;
            //string[] terms = lucandraDocument.IndexedTerms.Where(t => t.FieldName == field).Select(t => t.TermText).ToArray();
            //int[] frequencies = lucandraDocument.IndexedTerms.Where(t => t.FieldName == field).Select(t => t.Frequency).ToArray();
            //return new LucandraTermVector(field, terms, frequencies);
        }

        [Obsolete("Not implemented in LucandraReader.", true)]
        public override void GetTermFreqVector(int docNumber, string field, TermVectorMapper mapper)
        {
            throw new NotImplementedException();
        }

        [Obsolete("Not implemented in LucandraReader.", true)]
        public override void GetTermFreqVector(int docNumber, TermVectorMapper mapper)
        {
            throw new NotImplementedException();
        }

        /// <summary>
        /// Gets the number of documents currently known to this reader.
        /// </summary>
        /// <returns>Returns the total number of documents known by this reader.</returns>
        public override int NumDocs()
        {
            return this.DocumentCache.Count;
        }

        /// <summary>
        /// Gets one greater than the largest possible document number.  This may be used to, e.g., determine how many elements to
        /// allocate to an array which will have an element for every document number in an index.
        /// </summary>
        /// <returns>Returns one greater than the largest possible document number.</returns>
        public override int MaxDoc()
        {
            return MaximumDocuments + 1;
        }

        /// <summary>
        /// Gets the stored fields of the specified document number.
        /// </summary>
        /// <param name="documentNumber">The document number as known by this reader.</param>
        /// <returns>Returns a Lucene Document containing the stored fields.</returns>
        public override Document Document(int documentNumber)
        {
            var lucandraDocument = this.GetAndCacheLucandraDocument(documentNumber);
            if (lucandraDocument == null)
                throw new ArgumentException("The specified document number was not found in the reader.");

            return lucandraDocument.GetDocument();
        }

        [Obsolete("This method is not implemented in Lucandra.  Use Document(int n) overload instead.", true)]
        public override Document Document(int documentNumber, FieldSelector fieldSelector)
        {
            throw new NotImplementedException();    
        }

        /// <summary>
        /// Gets whether or not a document has been deleted.  LucandraReader will always return false for calls to this method.
        /// </summary>
        /// <param name="n">The document number as known by this reader.</param>
        /// <returns>Always returns false.</returns>
        public override bool IsDeleted(int n)
        {
            return false;
        }

        /// <summary>
        /// Gets whether or not the index has been optimized.  LucandraReader will always return true for calls to this method as optimizations are not possible/required.
        /// </summary>
        /// <returns>Always returns true.</returns>
        public override bool IsOptimized()
        {
            return true;
        }

        /// <summary>
        /// Gets whether or not any documents have been deleted.  LucandraReader will always return false for calls to this method.
        /// </summary>
        /// <returns>Always returns false.</returns>
        public override bool HasDeletions()
        {
            return false;
        }

        /// <summary>
        /// Gets whether or not there are norms stored for the specified field.
        /// </summary>
        /// <param name="field">The name of the field.</param>
        /// <returns>True if there are any known norms stored for this field, otherwise false.</returns>
        public override bool HasNorms(string field)
        {
            return (this.Norms(field) != null);
        }

        /// <summary>
        /// Gets the byte-encoded normalization factor for the named field of every document.  This is used by the search
        /// code to score documents.
        /// </summary>
        /// <param name="field">The name of the field.</param>
        /// <returns>Returns the byte-encoded normalization factor for the named field of every document.</returns>
        public override byte[] Norms(string field)
        {
            byte[] norms = null;
            this.FieldNorms.TryGetValue(field, out norms);
            return norms;
        }

        /// <summary>
        /// Reads the byte-encoded normalization factor for the named field of every document into the provided array (bytes).
        /// This is used by the search code to score documents.
        /// </summary>
        /// <param name="field">The name of the field.</param>
        /// <param name="bytes">The byte array into which the byte-encoded normalization factors should be copied.</param>
        /// <param name="offset">The offset at which to begin copying norms.</param>
        public override void Norms(string field, byte[] bytes, int offset)
        {
            byte[] norms;
            this.FieldNorms.TryGetValue(field, out norms);
            Array.Copy(norms, offset, bytes, 0, (bytes.Length > norms.Length ? norms.Length : bytes.Length));
        }

        /// <summary>
        /// Expert: Resets the normalization factor for the named field of the document.  The norm represents the product
        /// of the field's boost and its length normalization.  Thus, to preserve length normalization values when resetting
        /// this, one should base the new value upon the old.  NOTE: If this field does not store norms, then this method
        /// call will silently do nothing.
        /// </summary>
        /// <param name="doc">The document number as known by this reader.</param>
        /// <param name="field">The name of the field.</param>
        /// <param name="value">The new normalization factor.</param>
        protected override void DoSetNorm(int doc, string field, byte value)
        {
            byte[] norms;
            this.FieldNorms.TryGetValue(field, out norms);
            if (norms == null || doc >= norms.Length)
                return;
            norms[doc] = value;
        }

        /// <summary>
        /// Gets an enumeration of all of the terms in the index.  The enumeration is ordered as returned by Cassandra.  If using
        /// an out-of-the-box *OrderPreservingPartition then each returned term should be greater than all that precede it in the enumeration.
        /// Note that after calling Terms(), TermEnum.Next() must be called on the resulting enumeration before calling other methods
        /// such as TermEnum.Term().
        /// </summary>
        /// <returns>Returns an enumeration of all of the terms in the index.</returns>
        public override TermEnum Terms()
        {
            return new LucandraTermEnum(this);
        }

        /// <summary>
        /// Gets an enumeration of all terms in the index starting at a given term.  If the given term does not exist, the enumeration
        /// is positioned at the first term greater than the supplied term.  The enumeration is ordered as returned by Cassandra.  If
        /// using an out-of-the-box *OrderPreservingPartitioner then each returned term should be greater than all that precede it in the
        /// enumeration.
        /// </summary>
        /// <param name="t">The term at which to position the enumeration.</param>
        /// <returns>Returns an enumeration of all terms in the index starting at the given term.</returns>
        public override TermEnum Terms(Term t)
        {
            return new LucandraTermEnum(this, t);
        }

        /// <summary>
        /// Gets the number of documents in the index containing the given term.
        /// </summary>
        /// <param name="t">The term.</param>
        /// <returns>Returns the number of documents containing the given term.</returns>
        public override int DocFreq(Term t)
        {
            var termEnum = new LucandraTermEnum(this, t);
            return termEnum.DocFreq();
        }

        /// <summary>
        /// Gets an un-positioned TermDocs enumerator.
        /// </summary>
        /// <returns>Returns an un-positioned TermDocs enumerator.</returns>
        public override TermDocs TermDocs()
        {
            return new LucandraTermDocs(this);
        }

        /// <summary>
        /// Gets an un-positioned TermPositions enumerator.
        /// </summary>
        /// <returns>Returns an un-positioned TermPositions enumerator.</returns>
        public override TermPositions TermPositions()
        {
            return new LucandraTermPositions(this);
        }

        [Obsolete("Not yet implemented in Lucandra.NET.", true)]
        protected override void DoDelete(int docNum)
        {
        }
        
        [Obsolete("Not yet implemented in Lucandra.NET.", true)]
        protected override void DoUndeleteAll()
        {
        }

        [Obsolete("Not yet implemented in Lucandra.NET.", true)]
        protected override void DoCommit(IDictionary<string, string> commitUserData)
        {
        }

        [Obsolete("This will be removed in 3.0, do not use.", true)]
        protected override void DoCommit()
        {
        }

        protected override void DoClose()
        {
        }

        public override IndexReader Reopen()
        {
            return this;
        }

        public override ICollection<string> GetFieldNames(IndexReader.FieldOption fldOption)
        {
            return new string[0];
        }

        #endregion

        #region Methods [internal]

        /// <summary>
        /// Gets the document number (as known by this reader) of a document given its document ID.
        /// </summary>
        /// <param name="documentID">The document ID, as stored in Cassandra, without the index name or any delimeters..</param>
        /// <returns></returns>
        internal int GetDocumentNumberFromDocumentID(string documentID)
        {
            return this.DocumentIDToDocumentNumberMap[documentID];
        }

        /// <summary>
        /// Gets a LucandraDocument given the document number as known by this reader.
        /// </summary>
        /// <param name="documentNumber">The document number as known by this reader.</param>
        /// <returns></returns>
        internal LucandraDocument GetLucandraDocument(int documentNumber)
        {
            return this.DocumentCache[documentNumber];
        }

        internal int AddDocumentMapping(LucandraTermInfo.LucandraTermInfoDocument document, string fieldName)
        {
            int documentNumber;
            if (!this.DocumentIDToDocumentNumberMap.TryGetValue(document.DocumentID, out documentNumber))
            {
                this.DocumentCounter++;
                documentNumber = this.DocumentCounter;

                if (documentNumber > MaximumDocuments)
                    throw new InvalidOperationException("Maximum document count reached.");

                this.DocumentNumberToDocumentIDMap.Add(documentNumber, document.DocumentID);
                this.DocumentIDToDocumentNumberMap.Add(document.DocumentID, documentNumber);

                byte? norm = document.NormalizationFactor;
                if (norm == null)
                    norm = Similarity.EncodeNorm(1.0F);

                byte[] norms = null;
                if (!this.FieldNorms.TryGetValue(fieldName, out norms))
                    norms = new byte[128];

                while (norms.Length <= documentNumber && norms.Length < MaximumDocuments)
                {
                    byte[] newNorms = new byte[(norms.Length * 2) < MaximumDocuments ? (norms.Length * 2) : (MaximumDocuments + 1)];
                    Array.Copy(norms, 0, newNorms, 0, norms.Length);
                    norms = newNorms;
                }

                norms[documentNumber] = norm.Value;

                this.FieldNorms[fieldName] = norms;
            }

            return documentNumber;
        }

        #endregion
        #region Methods [private]

        private Dictionary<int, LucandraDocument> GetAndCacheLucandraDocuments(IEnumerable<int> documentNumbers)
        {
            var result = new Dictionary<int, LucandraDocument>();
            var searchDocumentKeys = new List<byte[]>();
            var documentKeyToDocumentIDMap = new Dictionary<string, string>();
            var documentIDToDocumentNumberMap = new Dictionary<string, int>();

            foreach (var documentNumber in documentNumbers)
            {
                LucandraDocument lucandraDocument;
                if (this.DocumentCache.TryGetValue(documentNumber, out lucandraDocument))
                    result.Add(documentNumber, lucandraDocument);
                else
                {
                    string documentID;
                    if (this.DocumentNumberToDocumentIDMap.TryGetValue(documentNumber, out documentID))
                    {
                        string documentKey = Settings.UseKeyspacePerIndex ? documentID : LucandraDocument.CreateDocumentKey(this.IndexName, documentID);
                        searchDocumentKeys.Add(Encoding.UTF8.GetBytes(documentKey));
                        documentKeyToDocumentIDMap[documentKey] = documentID;
                        documentIDToDocumentNumberMap[documentID] = documentNumber;
                    }
                }
            }

            // Create the column parent for column family [Documents].
            var columnParent = CassandraHelper.CreateColumnParent(Settings.DocumentsColumnFamily);
            // Create the slice range.  We only want to return two columns ('storedFields' and 'indexedTerms').
            var sliceRange = CassandraHelper.CreateSliceRange(new byte[0], new byte[0], false, 100);
            // Create the slice predicate.
            var slicePredicate = CassandraHelper.CreateSlicePredicate(sliceRange);

            var slices = new Dictionary<byte[], List<ColumnOrSuperColumn>>();

            // Get a reference to the client.  If a search context exists then use its client, otherwise borrow one from the pool.
            using (var client = CassandraConnectionPool.Borrow())
            {
                client.SetKeyspace(this.Keyspace);
                slices = client.InnerClient.multiget_slice(searchDocumentKeys, columnParent, slicePredicate, ConsistencyLevel.ONE);
            }

            if (slices.Count == 0)
                return result;

            foreach (byte[] key in slices.Keys)
            {
                string documentKey = Encoding.UTF8.GetString(key);
                var slice = slices[key];
                string documentID = documentKeyToDocumentIDMap[documentKey];
                int documentNumber = documentIDToDocumentNumberMap[documentID];

                // Create a new Lucandra document that we'll build upon.
                var lucandraDocument = new LucandraDocument(this.IndexName, documentID);

                // Iterate through all of the super columns in this key slice (there really only should be 'storedFields' and 'indexedTerms').
                foreach (var superColumn in slice)
                {
                    string superColumnName = Encoding.UTF8.GetString(superColumn.Super_column.Name);

                    // Iterate through all of the columns in the supercolumn.  This will be either the stored fields column "sf" or
                    // the indexed terms column "it".  Depending on the column, either a stored field or an indexed term is added
                    // to the LucandraDocument.
                    foreach (var col in superColumn.Super_column.Columns)
                    {
                        if (superColumnName == Settings.StoredFieldsColumn)
                        {
                            var storedField = LucandraDocument.StoredField.FromCassandraKeyValue(
                                Encoding.UTF8.GetString(col.Name), col.Value);
                            lucandraDocument.AddStoredField(storedField);
                        }
                        else if (superColumnName == Settings.IndexedTermsColumn)
                        {
                            var indexedTerm = LucandraDocument.IndexedTerm.FromCassandraKeyValue(
                                Encoding.UTF8.GetString(col.Name), col.Value);
                            lucandraDocument.AddIndexedTerm(indexedTerm);
                        }
                    }
                }

                // Add the document to the cache.
                this.DocumentCache[documentNumber] = lucandraDocument;

                result[documentNumber] = lucandraDocument;
            }

            return result;
        }

        private LucandraDocument GetAndCacheLucandraDocument(int documentNumber)
        {
            LucandraDocument lucandraDocument;
            if (this.DocumentCache.TryGetValue(documentNumber, out lucandraDocument))
                return lucandraDocument;

            string documentID;
            if (!this.DocumentNumberToDocumentIDMap.TryGetValue(documentNumber, out documentID))
                return null;

            //string documentKey = this.IndexName + Settings.Delimeter + documentID;
            //string documentKey = LucandraDocument.CreateDocumentKey(this.IndexName, documentID);
            string documentKey = Settings.UseKeyspacePerIndex ? documentID : LucandraDocument.CreateDocumentKey(this.IndexName, documentID);

            // Create the column parent for column family [Documents].
            var columnParent = CassandraHelper.CreateColumnParent(Settings.DocumentsColumnFamily);
            // Create the slice range.  We only want to return two columns ('storedFields' and 'indexedTerms').
            var sliceRange = CassandraHelper.CreateSliceRange(new byte[0], new byte[0], false, 100);
            // Create the slice predicate.
            var slicePredicate = CassandraHelper.CreateSlicePredicate(sliceRange);

            // Execute the search against Cassandra.
            List<ColumnOrSuperColumn> slice;

            // Get a reference to the client.  If a search context exists then use its client, otherwise borrow one from the pool.
            using (var client = CassandraConnectionPool.Borrow())
            {
                client.SetKeyspace(this.Keyspace);
                slice = client.InnerClient.get_slice(Encoding.UTF8.GetBytes(documentKey), columnParent, slicePredicate, ConsistencyLevel.ONE);
            }


            // If nothing was returned, return null - the document doesn't exist.
            if (slice == null)
                return null;

            // Create a new Lucandra document that we'll build upon.
            lucandraDocument = new LucandraDocument(this.IndexName, documentID);

            // Iterate through all of the super columns in this key slice (there really only should be 'storedFields' and 'indexedTerms').
            foreach (var superColumn in slice)
            {
                string superColumnName = Encoding.UTF8.GetString(superColumn.Super_column.Name);

                // Iterate through all of the columns in the supercolumn.  This will be either the stored fields column "sf" or
                // the indexed terms column "it".  Depending on the column, either a stored field or an indexed term is added
                // to the LucandraDocument.
                foreach (var col in superColumn.Super_column.Columns)
                {
                    if (superColumnName == Settings.StoredFieldsColumn)
                    {
                        var storedField = LucandraDocument.StoredField.FromCassandraKeyValue(
                            Encoding.UTF8.GetString(col.Name), col.Value);
                        lucandraDocument.AddStoredField(storedField);
                    }
                    else if (superColumnName == Settings.IndexedTermsColumn)
                    {
                        var indexedTerm = LucandraDocument.IndexedTerm.FromCassandraKeyValue(
                            Encoding.UTF8.GetString(col.Name), col.Value);
                        lucandraDocument.AddIndexedTerm(indexedTerm);
                    }
                }
            }

            // Add the document to the cache.
            this.DocumentCache[documentNumber] = lucandraDocument;

            return lucandraDocument;
        }

        #endregion
        #region Methods [public]

        public Document[] Documents(IEnumerable<int> documentNumbers)
        {
            var docs = this.GetAndCacheLucandraDocuments(documentNumbers);
            return docs .Values
                .Select(d => d.GetDocument())
                .ToArray();
        }

        #endregion
    }
}
